%%%-------------------------------------------------------------------
%%% File    : pipe_zip.erl
%%% Author  :  <vjache>
%%% Description : 
%%%
%%% Created : 22 Jul 2009 by  <vjache>
%%%-------------------------------------------------------------------
-module(pipe_pzip).

-include_lib("eunit/include/eunit.hrl").

-include("../../include/pipe.hrl").
-include("../../include/zip.hrl").

%% API
-export([new/2,
		 new/3,
		 open/1,
		 push/2,
		 flush/1,
		 close/1,
		 zip_list/2]).

%%====================================================================
%% API
%%====================================================================

%%--------------------------------------------------------------------
%% Function: new(BuffSize, NextPipe) -> pipe()
%% Description: Create pipe that compresses items pushed.  Items 
%%				accumulated to list with length BuffSize then list 
%%				compressed to so called zipped list tuple and sent to 
%%				NextPipe. Note that list to be zipped is in reversed 
%%				order by construction, that is if order of pushing 
%%				does matter, after unzip, the reverse of list must be done.
%%--------------------------------------------------------------------
new(BuffSize, NextPipe) when is_integer(BuffSize),
							 BuffSize>0,
							 is_record(NextPipe, pipe) ->
	pipe:pipe_from_module(?MODULE, {relax_order,BuffSize}, NextPipe).
%%--------------------------------------------------------------------
%% Function: new(ensure_order,BuffSize, NextPipe) -> pipe()
%% Description: The same as new/2 but reverses lists before zip operation.
%%--------------------------------------------------------------------
new(ensure_order,BuffSize, NextPipe) when is_integer(BuffSize),
							 BuffSize>0,
							 is_record(NextPipe, pipe) ->
	pipe:pipe_from_module(?MODULE, {ensure_order,BuffSize}, NextPipe).

open({{Reverse,BuffSize},NextPipe}) -> %%PipeState
    {[[]],Reverse,BuffSize,erlang:system_info(schedulers),pipe:open(NextPipe)}.
push({_Buff,_Reverse,_BuffSize,_SchedFree=0,_NextPipeState}=S,Item) ->
	push(flush(S),Item);
push({[Buff|Tail]=_Chunks,Reverse,_BuffSize=0,SchedFree,NextPipeState},Item) ->
	push({[[],Buff|Tail],Reverse,length(Buff),SchedFree-1,NextPipeState},Item);
push({[Buff|Tail]=_Chunks,Reverse,BuffFreeSize,SchedFree,NextPipeState},Item) ->
	{[[Item|Buff]|Tail],Reverse,BuffFreeSize-1,SchedFree,NextPipeState}.
flush({[Buff|_Tail]=Chunks,Reverse,BuffFreeSize,SchedFree,NextPipeState}) ->
	NextPipeState1=lists:foldr(
	  fun(ZipList,PS)->
			  pipe:push(PS,ZipList)
	  end,
	  NextPipeState,
	  pmap(fun(Chunk)->
				   zip_list(Chunk,Reverse)
		   end,
		   Chunks)),
	{[[]],Reverse,length(Buff)+BuffFreeSize,SchedFree+length(Chunks),NextPipeState1}.
close({[[]],_Reverse,_BuffFreeSize,_SchedFree,NextPipeState}) ->
	pipe:close(NextPipeState),
	ok.
%%====================================================================
%% Internal functions
%%====================================================================
zip_list(List,ensure_order) ->
	ListLength=length(List),
	?ZIP_LIST_REVERSED(ListLength,zlib:zip(term_to_binary(lists:reverse(List))));
zip_list(List,relax_order) ->
	ListLength=length(List),
	?ZIP_LIST_NOTREVERSED(ListLength,zlib:zip(term_to_binary(List))).

pmap(Fun,List) when is_function(Fun, 1)->
	Pids=pmap(Fun,List,[]),
	[receive
		 {Pid,{error,_Reason,_At}=Err} ->
			 throw(Err);
		 {Pid,{Result}}->
			 Result
	 end||Pid<-Pids].

pmap(_Fun,[],Pids) ->
	Pids;
pmap(Fun,[Item|Tail]=_List,Pids) ->
	MasterPid=self(),
	Pid=spawn_link(
	  fun()->
			  try Fun(Item) of
				  Result ->
					  MasterPid ! {self(),{Result}}
			  catch
				  _:Reason ->
					  MasterPid ! 
						  {self(),{error,Reason,erlang:get_stacktrace()}}
			  end
	  end),
	pmap(Fun,Tail,[Pid|Pids]).

%%====================================================================
%% eunit test functions
%%====================================================================

zip_test() ->
	file:delete("test_asis.LOG"),
%% 	file:delete("test_zip.LOG"),
%% 	Big=3000000,
%% 	TestData=[{X,Y,Z}||X<-lists:seq(1,100),Y<-lists:seq(101,130),Z<-lists:seq(Big,Big+100)],
%% 	Pipe=pipe_disk_logger:new([{name,test_asis}]),
%%  	Pump=pump:new(iter:from_list(TestData), Pipe),
%% 	pump:init(Pump),
	Pump1=pump:new(iter_disk_log:new([{name,test_asis}]),
				   new(1000,pipe_disk_logger:new([{name,test_zip}]))),
	pump:init(Pump1)
%% ,
%% 	?assertEqual(true,filelib:is_regular("test_asis.LOG")),
%% 	?assertEqual(true,filelib:is_regular("test_zip.LOG")),
%% 	TestData
.

zip_unzip_test() ->
	TestData=zip_test(),
	TestData1=iter:to_list(iter_unzip:new(ensure_order,iter_disk_log:new([{name,test_zip}]))),
	?assertEqual(TestData,TestData1).

zip_unzip_1_test() ->
	file:delete("test_asis.LOG"),
	file:delete("test_zip.LOG"),
	Big=3000000,
	TestData=[{X,Y,Z}||X<-lists:seq(1,100),Y<-lists:seq(101,130),Z<-lists:seq(Big,Big+100)],
	Pipe=pipe_disk_logger:new([{name,test_asis}]),
	Pump=pump:new(iter:from_list(TestData), Pipe),
	pump:init(Pump),
	Pump1=pump:new(iter_disk_log:new([{name,test_asis}]),
				   new(ensure_order,1000,pipe_disk_logger:new([{name,test_zip}]))),
	pump:init(Pump1),
	?assertEqual(true,filelib:is_regular("test_asis.LOG")),
	?assertEqual(true,filelib:is_regular("test_zip.LOG")),
	TestData1=iter:to_list(iter_unzip:new(iter_disk_log:new([{name,test_zip}]))),
	?assertEqual(TestData,TestData1).
	
